package com.yyx.demo.publishSubscribe;

import com.yyx.demo.queue.ZKDistributedQueue;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.ACL;

import java.util.List;

public class SubscribeWatcher implements Watcher {

    private String dir=null;

    private final String sessionId;

    private List<ACL> acl = ZooDefs.Ids.OPEN_ACL_UNSAFE;

    private ZooKeeper zooKeeper;

    private SequenceZnode sequenceZnode;

    public SubscribeWatcher(ZooKeeper zooKeeper, List<ACL> acl, String sessionId) {
        this.zooKeeper=zooKeeper;
        if (acl != null) {
            this.acl = acl;
        }
        this.sessionId=sessionId;
        this.sequenceZnode= new SequenceZnode(zooKeeper,acl);
    }

    public String getDir() {
        return dir;
    }

    public void setDir(String dir) {
        this.dir = dir;
    }

    @Override
    public void process(WatchedEvent event) {
        //监听到创建节点
        if(event.getType().getIntValue()==Event.EventType.NodeChildrenChanged.getIntValue()){
            try {
                String path="/"+sessionId+dir;
                //消费事件，并存入分布式队列
                System.out.println("收到通知");
                ZKDistributedQueue zkDistributedQueue = new ZKDistributedQueue(zooKeeper,path,acl);
                zkDistributedQueue.offerSync(sequenceZnode.take(dir));
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };

    }

}
